package com.tmt.future;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import com.swak.reactivex.threads.SwakThreadFactory;

import reactor.core.publisher.Mono;

/**
 * 
 * 测试结果， then run when 等不能切换线程的api 有可能会在调用者线程中执行代码，
 * 如果调用时前面已经有结果了，则会直接在调用线程中执行相应的代码。
 * 
 * 所以如果是耗时间的操作则不要放在 then run when 中执行。
 * 
 * @author lifeng
 */
public class TestFuture2 {

	// /**
	// * 进行变换 supplyAsync -- thenApplyAsync
	// */
	// @Test
	// public void thenApply() {
	// String result = CompletableFuture.supplyAsync(() -> "hello").thenApplyAsync(s
	// -> s + " world").join();
	// System.out.println(result);
	// }
	//
	// /**
	// * 进行消耗 supplyAsync -- thenAcceptAsync
	// */
	// @Test
	// public void thenAccept() {
	// CompletableFuture.supplyAsync(() -> "hello").thenAcceptAsync(s ->
	// System.out.println(s + " world2"));
	// }
	//
	// /**
	// * 对上一步的计算结果不关心，执行下一个操作。
	// * supplyAsync -- thenRunAsync
	// */
	// @Test
	// public void thenRun() {
	// CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(2000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "hello";
	// }).thenRunAsync(() -> System.out.println("hello world"));
	// }
	//
	// /**
	// * 结合两个CompletionStage的结果，进行转化后返回
	// */
	// @Test
	// public void thenCombine() {
	// String result = CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(2000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "hello1";
	// }).thenCombine(CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "world1";
	// }), (s1, s2) -> s1 + " " + s2).join();
	// System.out.println(result);
	// }
	//
	// /**
	// * 结合两个CompletionStage的结果，进行消耗
	// */
	// @Test
	// public void thenAcceptBoth() {
	// CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(2000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "hello2";
	// }).thenAcceptBoth(CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "world2";
	// }), (s1, s2) -> System.out.println(s1 + " " + s2));
	// while (true){}
	// }
	//
	// @Test
	// public void runAfterBoth(){
	// CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(2000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "s1";
	// }).runAfterBothAsync(CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "s2";
	// }), () -> System.out.println("hello world"));
	// while (true){}
	// }
	//
	// /**
	// * 两个CompletionStage，谁计算的快，我就用那个CompletionStage的结果进行下一步的转化操作。
	// * apply 有返回值
	// */
	// @Test
	// public void applyToEither() {
	// String result = CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "s1";
	// }).applyToEither(CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(2000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "hello world";
	// }), s -> s).join();
	// System.out.println(result);
	// }
	//
	// /**
	// * 两个CompletionStage，谁计算的快，我就用那个CompletionStage的结果进行下一步的消耗操作。
	// * accept 无返回值
	// */
	// @Test
	// public void acceptEither() {
	// CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "s1";
	// }).acceptEither(CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(2000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "hello world";
	// }), System.out::println);
	// while (true){}
	// }
	//
	// /**
	// * 两个CompletionStage，任何一个完成了都会执行下一步的操作（Runnable）。
	// * run 是不关心计算结果
	// */
	// @Test
	// public void runAfterEither() {
	// CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "s1";
	// }).runAfterEither(CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(2000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// return "s2";
	// }), () -> System.out.println("hello world"));
	// while (true) {
	// }
	// }

	// /**
	// * exceptionally 捕获异常
	// */
	// @Test
	// public void exceptionally() {
	// String result = CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// if (1 == 1) {
	// throw new RuntimeException("测试一下异常情况");
	// }
	// return "s1";
	// }).exceptionally(e -> {
	// System.out.println(e.getMessage());
	// return "hello world";
	// }).join();
	// System.out.println(result);
	// }

	// /**
	// * 当运行完成时，对结果的记录。
	// */
	// @Test
	// public void whenComplete() {
	// String result = CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// if (1 == 1) {
	// throw new RuntimeException("测试一下异常情况");
	// }
	// return "s1";
	// }).whenComplete((s, t) -> {
	// System.out.println(s); //获取不到
	// System.out.println(t.getMessage());
	// }).exceptionally(e -> {
	// //System.out.println(e.getMessage());
	// return "hello world";
	// }).join();
	// System.out.println(result);
	// }

	// /**
	// *
	// */
	// @Test
	// public void handle() {
	// String result = CompletableFuture.supplyAsync(() -> {
	// try {
	// Thread.sleep(3000);
	// } catch (InterruptedException e) {
	// e.printStackTrace();
	// }
	// //出现异常
	// if (1 == 1) {
	// throw new RuntimeException("测试一下异常情况");
	// }
	// return "s1";
	// }).handle((s, t) -> {
	// if (t != null) {
	// return "hello world";
	// }
	// return s;
	// }).join();
	// System.out.println(result);
	// }
	//
	// public static CompletableFuture<String> calculate(String input) {
	// ExecutorService executor = Executors.newFixedThreadPool(5);
	// CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
	// System.out.println(input);
	// return input + "---" + input.length();
	// }, executor);
	// return future;
	// }
	//
	// @Test
	// public void testThenCompose() throws InterruptedException, ExecutionException
	// {
	// ExecutorService executor = Executors.newFixedThreadPool(5);
	// CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
	// return "zero";
	// }, executor);
	//
	// CompletableFuture<CompletableFuture<String>> f4 =
	// f1.thenApply(TestFuture2::calculate);
	// System.out.println("f4.get:"+f4.get().get());
	//
	// // 类似flatMap
	// CompletableFuture<String> f5 = f1.thenCompose(TestFuture2::calculate);
	// System.out.println("f5.get:"+f5.get());
	//
	// System.out.println(f1.join());
	// }

	private void sleep() {
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * CompletableFuture 可能会在调用者的线程中执行 thenApply，所以如果是耗时间的操作最好 使用 thenApplyAsync
	 * 且明确指定所使用的线程池
	 * 
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	@Test
	public void testCompletableFuture() throws InterruptedException, ExecutionException {
		CountDownLatch latch = new CountDownLatch(1);
		ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2,
				new SwakThreadFactory("李锋-", true, new AtomicInteger(0)));
		CompletableFuture<Void> f = CompletableFuture.supplyAsync(() -> {
			System.out.println("当前线程1" + Thread.currentThread().getName());
			sleep();
			return "1";
		}, executor).whenComplete((v, e) -> {
			System.out.println("当前线程3" + Thread.currentThread().getName());
		}).thenAccept((s) -> {
			System.out.println("当前线程2" + Thread.currentThread().getName());
		}).whenComplete((v, e) -> {
			System.out.println("当前线程4" + Thread.currentThread().getName());
		});
		Mono.fromCompletionStage(f).doOnSuccess(s -> {
			System.out.println("success 1");
		}).doOnSuccess(s -> {
			System.out.println("success 2");
		}).subscribe(new Subscriber<Void>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(Long.MAX_VALUE);
			}

			@Override
			public void onNext(Void t) {
				System.out.println("onNext");
			}

			@Override
			public void onError(Throwable t) {
				System.out.println("当前线程final" + Thread.currentThread().getName());
				System.out.println("onError");
				latch.countDown();
			}

			@Override
			public void onComplete() {
				System.out.println("onComplete");
				latch.countDown();
			}
		});
		latch.await();
	}
}
